package netty.queue;

import com.lmax.disruptor.BusySpinWaitStrategy;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import netty.tcp.TCPClient;

public class RBQ {

    Disruptor<ZetasEvent> disruptor;
    RingBuffer<ZetasEvent> ringBuffer;

    private RBQ(TCPClient tcpClient, int ringBufferSize) {
        StringEventFactory eventFactory = new StringEventFactory();
        disruptor = new Disruptor<>(eventFactory, ringBufferSize, new ZetasThreadFactory(),ProducerType.SINGLE, new BusySpinWaitStrategy());
        disruptor.handleEventsWith(new StringEventHandler(tcpClient));

        disruptor.start();


        ringBuffer = disruptor.getRingBuffer();
    }

    public static RBQ create(TCPClient tcpClient, int ringBufferSize) {
        return new RBQ(tcpClient, ringBufferSize);
    }

    public void sendMsg(byte[] msg) {
        ringBuffer.publishEvent((event, sequence) -> event.setMsg(msg).setNanaTime());
    }

}
